package com.shujia.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo03MySink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val stuDS: DataStream[String] = env.readTextFile("Flink/data/stu/students.txt")

    stuDS.addSink(new MySink)

    env.execute()

  }

}

// 自定义Sink将数据写入MySQL
class MySink extends RichSinkFunction[String] {
  var conn: Connection = _

  override def open(parameters: Configuration): Unit = {
    // 建立连接
    Class.forName("com.mysql.jdbc.Driver")
    conn = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
  }

  override def close(): Unit = {
    conn.close()

  }

  // 每条数据会执行一次invoke方法
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
    val preSt: PreparedStatement = conn.prepareStatement("insert into student_like values(?,?,?,?,?)")

    val stuArr: Array[String] = value.split(",")

    preSt.setInt(1, stuArr(0).toInt)
    preSt.setString(2, stuArr(1))
    preSt.setInt(3, stuArr(2).toInt)
    preSt.setString(4, stuArr(3))
    preSt.setString(5, stuArr(4))

    preSt.execute()


  }
}
